/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.examples.streaming;

import java.io.Serializable;
import java.util.List;
import java.util.Random;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cache.query.annotations.QuerySqlField;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.examples.ExampleNodeStartup;
import org.apache.ignite.examples.ExamplesUtils;
import org.apache.ignite.stream.StreamVisitor;

/**
 * Ignite流处理示例：使用StreamVisitor处理实时市场数据流。
 *
 * 这个示例演示了如何使用Ignite的流处理功能来实时处理和分析数据流。
 * 具体场景是模拟金融市场的股票价格数据处理：
 *
 * 核心概念：
 * 1. **IgniteDataStreamer**: 高性能数据流加载器，支持海量数据的快速写入
 * 2. **StreamVisitor**: 流数据访问器，在数据写入时执行自定义逻辑
 * 3. **数据并置**: 确保相关数据在同一个节点上，避免网络开销
 * 4. **实时分析**: 在数据流入时实时计算和更新统计信息
 *
 * 处理流程：
 * 1. 创建市场数据流接收器
 * 2. 使用StreamVisitor处理每个流入的数据点
 * 3. 实时更新金融工具的统计信息（开盘价、最新价）
 * 4. 执行SQL查询找出表现最佳的金融工具
 *
 * 运行要求：
 * 1. 启动几个服务器节点（使用ExampleNodeStartup）
 * 2. 运行此示例作为客户端节点
 *
 * 学习要点：
 * - 流式数据处理的基本模式
 * - StreamVisitor的使用方法
 * - 数据并置的重要性
 * - 实时分析的实现方式
 */
public class StreamVisitorExample {
    /** 随机数生成器：用于模拟市场价格波动 */
    private static final Random RAND = new Random();

    /** 金融工具列表：包含10家知名科技公司的股票代码 */
    private static final String[] INSTRUMENTS = {"IBM", "GOOG", "MSFT", "GE", "EBAY", "YHOO", "ORCL", "CSCO", "AMZN", "RHT"};

    /** 初始价格列表：对应上述股票代码的起始价格 */
    private static final double[] INITIAL_PRICES = {194.9, 893.49, 34.21, 23.24, 57.93, 45.03, 44.41, 28.44, 378.49, 69.50};

    /**
     * 程序入口点：演示完整的流处理流程。
     *
     * @param args 命令行参数（不需要任何参数）
     * @throws Exception 如果执行过程中发生错误
     */
    public static void main(String[] args) throws Exception {
        // 将此集群成员设置为客户端模式，避免存储数据
        Ignition.setClientMode(true);

        try (Ignite ignite = Ignition.start("examples/config/example-ignite.xml")) {
            // 检查是否有服务器节点可用
            if (!ExamplesUtils.hasServerNodes(ignite))
                return;

            // 市场数据缓存配置：存储实时价格数据（实际上不会被填充）
            CacheConfiguration<String, Double> mktDataCfg = new CacheConfiguration<>("marketTicks");

            // 金融工具缓存配置：存储股票的统计信息
            CacheConfiguration<String, Instrument> instCfg = new CacheConfiguration<>("instCache");

            // 为金融工具缓存设置索引，支持SQL查询
            // 注意：Instrument类中的字段使用@QuerySqlField注解进行索引
            instCfg.setIndexedTypes(String.class, Instrument.class);

            // 使用try-with-resources确保缓存自动关闭
            try (
                IgniteCache<String, Double> mktCache = ignite.getOrCreateCache(mktDataCfg);
                IgniteCache<String, Instrument> instCache = ignite.getOrCreateCache(instCfg)
            ) {
                // 创建数据流处理器：用于高性能批量写入
                try (IgniteDataStreamer<String, Double> mktStmr = ignite.dataStreamer(mktCache.getName())) {
                    // 设置StreamVisitor：在数据流入时执行自定义处理逻辑
                    // 重要概念：
                    // 1. 我们接收市场数据，但不填充'mktCache'（它保持为空）
                    // 2. 而是更新'instCache'中的金融工具信息
                    // 3. 由于'instCache'和'mktCache'使用相同的键，更新是并置的（在同一个节点）
                    mktStmr.receiver(StreamVisitor.from((cache, e) -> {
                        String symbol = e.getKey();  // 股票代码
                        Double tick = e.getValue();  // 最新价格

                        // 从缓存中获取对应的金融工具
                        Instrument inst = instCache.get(symbol);

                        // 如果不存在，创建新的金融工具对象
                        if (inst == null)
                            inst = new Instrument(symbol);

                        // 不填充市场缓存，因为不用于查询
                        // 根据最新市场价格更新缓存的金融工具
                        inst.update(tick);

                        // 将更新后的金融工具放回缓存
                        instCache.put(symbol, inst);
                    }));

                    // 向系统流式传输1000万个市场数据点
                    for (int i = 1; i <= 10_000_000; i++) {
                        int idx = RAND.nextInt(INSTRUMENTS.length);

                        // 使用高斯分布确保接近0的数字具有更高的概率
                        // 这模拟了实际市场价格围绕均值波动的情况
                        double price = round2(INITIAL_PRICES[idx] + RAND.nextGaussian());

                        // 将数据添加到流中，触发StreamVisitor处理
                        mktStmr.addData(INSTRUMENTS[idx], price);

                        // 每50万个数据点输出一次进度
                        if (i % 500_000 == 0)
                            System.out.println("Number of tuples streamed into Ignite: " + i);
                    }
                }

                // 查询表现最佳的3个金融工具
                // 计算公式：(最新价格 - 开盘价格) = 涨跌幅
                SqlFieldsQuery top3qry = new SqlFieldsQuery(
                    "select symbol, (latest - open) from Instrument order by (latest - open) desc limit 3");

                // 执行SQL查询
                List<List<?>> top3 = instCache.query(top3qry).getAll();

                System.out.println("Top performing financial instruments: ");

                // 打印查询结果（使用工具类美化输出格式）
                ExamplesUtils.printQueryResults(top3);
            }
            finally {
                // 分布式缓存只能通过#destroyCache()调用从集群中移除
                // 清理演示创建的缓存，避免影响后续测试
                ignite.destroyCache(mktDataCfg.getName());
                ignite.destroyCache(instCfg.getName());
            }
        }
    }

    /**
     * 将double值四舍五入到两位小数。
     *
     * 这个方法用于模拟金融价格的标准格式（两位小数）。
     * 金融价格通常保留2位小数，表示"分"级别的精度。
     *
     * @param val 要四舍五入的值
     * @return 四舍五入到两位小数的double值
     */
    private static double round2(double val) {
        return Math.floor(100 * val + 0.5) / 100;
    }

    /**
     * 金融工具实体类：表示股票或其他金融产品的统计信息。
     *
     * 这个类展示了如何在Ignite中创建可查询的实体：
     * 1. 实现Serializable接口，支持序列化
     * 2. 使用@QuerySqlField注解创建SQL字段和索引
     * 3. 提供实时更新的方法
     * 4. 跟踪关键价格指标（开盘价、最新价）
     *
     * 数据结构说明：
     * - symbol: 股票代码（不可变）
     * - open: 开盘价（首次收到价格时设置）
     * - latest: 最新价格（实时更新）
     */
    public static class Instrument implements Serializable {
        /** 股票代码：唯一标识符，使用@QuerySqlField创建索引 */
        @QuerySqlField(index = true)
        private final String symbol;

        /** 开盘价格：第一次收到的价格，使用@QuerySqlField创建索引 */
        @QuerySqlField(index = true)
        private double open;

        /** 最新价格：最近一次收到的价格，使用@QuerySqlField创建索引 */
        @QuerySqlField(index = true)
        private double latest;

        /**
         * 构造函数：创建新的金融工具实例。
         *
         * @param symbol 股票代码（如"IBM"、"GOOG"等）
         */
        public Instrument(String symbol) {
            this.symbol = symbol;
        }

        /**
         * 根据最新的市场价格更新金融工具信息。
         *
         * 这个方法实现了实时更新的逻辑：
         * 1. 如果是第一次收到价格，设置为开盘价
         * 2. 始终更新最新价格
         * 3. 为后续的涨跌计算提供数据基础
         *
         * 更新策略：
         * - 开盘价：只在第一次更新时设置，之后保持不变
         * - 最新价：每次调用都会更新
         *
         * @param price 最新的市场价格
         */
    }
}
